package com.bbx.flink.demo.trigger;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

@Slf4j
public class CustomTriggerDemo {

    public static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.10.131", 10003)
                .map((MapFunction<String, Integer>)value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(1))
                .trigger(new Trigger<Integer, TimeWindow>() {
                    private static final long serialVersionUID = 7419116843788513049L;

                    @Override
                    public TriggerResult onElement(Integer element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~onElement方法,element[{}]",element);
                        return TriggerResult.FIRE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~onProcessingTime方法，time[{}]",time);
                        return TriggerResult.FIRE_AND_PURGE;
                    }

                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
                        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~clear 方法[{}]");

                    }
                    //此处canMerge    onMerge 方法是可选的重写方法，
                    @Override
                    public boolean canMerge() {
                        return super.canMerge();
                    }

                    @Override
                    public void onMerge(TimeWindow window, OnMergeContext ctx) throws Exception {
                        super.onMerge(window, ctx);
                    }
                }).sum(0)
                .print();

        env.execute();
    }
}
